/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.rocketmq.client.latency;

import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.common.message.MessageQueue;

public class MQFaultStrategy {
    private final static InternalLogger log = ClientLogger.getLog();
    //延迟隔离容错的具体实现(broker列表)
    private final LatencyFaultTolerance<String> latencyFaultTolerance = new LatencyFaultToleranceImpl();
    //生产者延迟隔离策略开关(默认不开启)
    private boolean sendLatencyFaultEnable = false;
    
    //rt耗费时间列表(rt耗时等级)
    private long[] latencyMax = {50L, 100L, 550L, 1000L, 2000L, 3000L, 15000L};
    //根据rt耗时等级，寻找broker失效时间(例如：生产者的rt耗时是1s,对应latencyMax的index=3,则broker的故障隔离时间就是notAvailableDuration[3]=60s)
    private long[] notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};

    public long[] getNotAvailableDuration() {
        return notAvailableDuration;
    }

    public void setNotAvailableDuration(final long[] notAvailableDuration) {
        this.notAvailableDuration = notAvailableDuration;
    }

    public long[] getLatencyMax() {
        return latencyMax;
    }

    public void setLatencyMax(final long[] latencyMax) {
        this.latencyMax = latencyMax;
    }

    public boolean isSendLatencyFaultEnable() {
        return sendLatencyFaultEnable;
    }

    public void setSendLatencyFaultEnable(final boolean sendLatencyFaultEnable) {
        this.sendLatencyFaultEnable = sendLatencyFaultEnable;
    }
    //选择生产者发送消息的分区queue(负载均衡策略 + 延迟隔离策略)
    public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
        //是否开启隔离策略(在producer启动时开启，默认未开始)，高可用环境建议打开
        if (this.sendLatencyFaultEnable) {
            try {
                //轮询(round-robin)获取被选中queue的index
                int index = tpInfo.getSendWhichQueue().getAndIncrement();
                //从index开始轮询循环遍历topic的分区，找到没有在隔离中的队列queue
                for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
                    int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
                    if (pos < 0)
                        pos = 0;
                    MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
                    //判断取到队列的broker是否在隔离中(延迟隔离策略以broker为单位，避免一个broker故障后消息重试做多次判断)
                    if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {
                        //如不在隔离列表中就直接返回
                        if (null == lastBrokerName || mq.getBrokerName().equals(lastBrokerName))
                            return mq;
                    }
                }
                //选次优broker：如果所有的队列都被隔离了，就从故障隔离列表中取出一个次优broker发送消息
                final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
                //获取选中broker的写队列（broker没有可写队列返回-1）
                int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
                if (writeQueueNums > 0) {
                    //轮询选择topic的分区队列queue
                    final MessageQueue mq = tpInfo.selectOneMessageQueue();
                    if (notBestBroker != null) {
                        mq.setBrokerName(notBestBroker);
                        mq.setQueueId(tpInfo.getSendWhichQueue().getAndIncrement() % writeQueueNums);
                    }
                    return mq;
                } else {
                    //无可写queue，就从故障隔离列表中移除
                    latencyFaultTolerance.remove(notBestBroker);
                }
            } catch (Exception e) {
                log.error("Error occurred when selecting message queue", e);
            }

            return tpInfo.selectOneMessageQueue();
        }
        //负载均衡：轮询(round-robin)获取发送消息的分区queue
        return tpInfo.selectOneMessageQueue(lastBrokerName);
    }
    
    //更新故障项：根据耗时 + isolation 决定要不要将该broker加入故障列表中
    public void updateFaultItem(final String brokerName, final long currentLatency, boolean isolation) {
        //开启延迟隔离容错机制
        if (this.sendLatencyFaultEnable) {
            //计算出隔离时间，默认隔离30s(根据耗时性能来选择故障隔离时间)
            long duration = computeNotAvailableDuration(isolation ? 30000 : currentLatency);
            this.latencyFaultTolerance.updateFaultItem(brokerName, currentLatency, duration);
        }
    }
    //根据rt时间来选择隔离时长(broker不可用时长，从隔离意义上来说，肯定是避免性能差的broker使用次数)
    private long computeNotAvailableDuration(final long currentLatency) {
        for (int i = latencyMax.length - 1; i >= 0; i--) {
            if (currentLatency >= latencyMax[i])
                return this.notAvailableDuration[i];
        }
        return 0;
    }
}
